package com.dctp.common.core;

import com.dctp.common.utils.UUIDUtil;
import com.jfinal.kit.Prop;
import com.jfinal.kit.PropKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.params.SetParams;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

public class RedisManager {

    private static final Logger logger = LoggerFactory.getLogger(RedisManager.class);
    private static RedisManager instance;
    private final JedisPool jedisPool;

    // 默认参数（可以用配置文件替换）
    private static final String HOST = "127.0.0.1";
    private static final int PORT = 6379;
    private static final int TIMEOUT = 2000; // ms
    private static String PASSWORD = "89a7ef4e3508425fb4bd687cc9ffa65b";
    private static final int DATABASE = 0;

    private RedisManager(String host, int port) {
        JedisPoolConfig config = new JedisPoolConfig();

        // ✅ 连接池配置（见第 2 节详细说明）
        config.setMaxTotal(600);       // 最大连接数
        config.setMaxIdle(100);        // 最大空闲连接
        config.setMinIdle(20);         // 最小空闲连接
        config.setTestOnBorrow(true);  // 取连接时测试
        config.setTestOnReturn(false);
        config.setTestWhileIdle(true);

        Prop p = PropKit.use("project.properties");
        String env = p.get("env","prod");
        if (!"prod".equals(env)) {
            PASSWORD = null;
            logger.info("当前环境:{}，将密码设置为NULL",env);
        }
        this.jedisPool = new JedisPool(config, host, port, TIMEOUT, PASSWORD, DATABASE);
    }

    public static void init(String host, int port) {
        synchronized (RedisManager.class) {
            if (instance != null) {
                instance.close();
            }
            instance = new RedisManager(host, port);
        }
    }

    public static RedisManager getInstance() {
        if (instance == null) {
            synchronized (RedisManager.class) {
                if (instance == null) {
                    instance = new RedisManager(HOST, PORT);
                }
            }
        }
        return instance;
    }






    /** 关闭连接池 */
    public void close() {
        if (jedisPool != null) {
            jedisPool.close();
        }
    }

    // ✅ Jedis 资源管理模板
    private <T> T execute(Function<Jedis, T> callback) {
        try (Jedis jedis = jedisPool.getResource()) {
            return callback.apply(jedis);
        } catch (Exception e) {
            logger.error("Redis 执行异常", e);
            throw e;
        }
    }


    /* ==================== 基础方法 ==================== */

    public Long getLong(String key) {
        return execute(jedis -> {
            String val = jedis.get(key);
            return (val != null) ? Long.valueOf(val) : null;
        });
    }

    public void setex(String key, long ttlSeconds, Object value) {
        execute(jedis -> jedis.setex(key, (int) ttlSeconds, String.valueOf(value)));
    }

    public void expireAt(String key, long unixTime) {
        execute(jedis -> jedis.expireAt(key, unixTime));
    }

    public Long incr(String key) {
        return execute(jedis -> jedis.incr(key));
    }

    public double incrByFloat(String key,double value) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.incrByFloat(key,value);
        }
    }

    public Long decrBy(String key, long value) {
        return execute(jedis -> jedis.decrBy(key, value));
    }

    // ------------------ 常用操作示例 ------------------

    public String get(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.get(key);
        }
    }

    public void set(String key, String value) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.set(key, value);
        }
    }

    public void expire(String key, int seconds) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.expire(key, seconds);
        }
    }


    public void setex(String key, String value, int seconds) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.setex(key, seconds, value);
        }
    }

    public boolean setnx(String key, String value) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.setnx(key, value) > 0;
        }
    }

    public boolean setIfAbsent(String key, String value, int expireSeconds) {
        try (Jedis jedis = jedisPool.getResource()) {
            SetParams params = new SetParams().nx().ex(expireSeconds);
            String result = jedis.set(key, value, params);
            return "OK".equals(result);
        }
    }

    public Long del(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.del(key);
        }
    }

    public Boolean exists(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.exists(key);
        }
    }

    // Hash 操作
    public String hget(String key, String field) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.hget(key, field);
        }
    }

    public void hset(String key, String field, String value) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.hset(key, field, value);
        }
    }

    public List<String> hmget(String key, String... fields) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.hmget(key, fields);
        }
    }

    public void hdel(String key, String field) {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.hdel(key, field);
        }
    }

    public Map<String, String> hgetAll(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.hgetAll(key);
        }
    }


    // ---- pipeline
    public void pipeline(java.util.function.Consumer<Pipeline> consumer) {
        try (Jedis jedis = jedisPool.getResource()) {
            Pipeline pipeline = jedis.pipelined();
            consumer.accept(pipeline);
            pipeline.sync();
        }
    }

    public <T> java.util.List<T> pipelineWithResult(
            java.util.function.Function<Pipeline, java.util.List<Response<T>>> fn) {
        try (Jedis jedis = jedisPool.getResource()) {
            Pipeline pipeline = jedis.pipelined();
            java.util.List<Response<T>> responses = fn.apply(pipeline);
            pipeline.sync();
            java.util.List<T> result = new java.util.ArrayList<>();
            for (Response<T> r : responses)
                result.add(r.get());
            return result;
        }
    }

    // ---- transaction
    public void transaction(java.util.function.Consumer<Transaction> consumer) {
        try (Jedis jedis = jedisPool.getResource()) {
            Transaction tx = jedis.multi();
            consumer.accept(tx);
            tx.exec();
        }
    }

    public java.util.List<Object> transactionWithResult(
            java.util.function.Function<Transaction, java.util.List<Response<?>>> fn) {
        try (Jedis jedis = jedisPool.getResource()) {
            Transaction tx = jedis.multi();
            java.util.List<Object> result = tx.exec();
            return result;
        }
    }

    // ---- pub/sub
    public Long publish(String channel, String message) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.publish(channel, message);
        }
    }

    public void subscribe(JedisPubSub listener, String... channels) {
        new Thread(() -> {
            while (true) {
                Jedis jedis = null;
                try {
                    jedis = jedisPool.getResource();
                    logger.info("✅ Redis 订阅启动: {}", Arrays.toString(channels));
                    jedis.subscribe(listener, channels); // 阻塞直到异常或取消
                } catch (Exception e) {
                    logger.error("❌ Redis subscribe 异常，5 秒后重试", e);
                    try { Thread.sleep(5000); } catch (InterruptedException ignored) {}
                } finally {
                    // 🔒 只有异常退出时才会执行这里
                    if (jedis != null) {
                        try { jedis.close(); } catch (Exception ignore) {}
                    }
                }
            }
        }, "redis-subscribe-" + channels[0]).start();
    }


    // Set
    public Long sadd(String key, String... members) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.sadd(key, members);
        }
    }

    public java.util.Set<String> smembers(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.smembers(key);
        }
    }

    // List
    public Long lpush(String key, String... values) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.lpush(key, values);
        }
    }

    public String lpop(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.lpop(key);
        }
    }


    public String blpop(String key, int timeoutSeconds) {
        try (Jedis jedis = jedisPool.getResource()) {
            List<String> result = jedis.blpop(timeoutSeconds, key);
            // result 可能为 null 或长度为2：[key, value]
            if (result != null && result.size() == 2) {
                return result.get(1); // value
            }
            return null;
        }
    }

    public String blpop(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            List<String> result = jedis.blpop(0, key); // 0 = 永久阻塞
            if (result != null && result.size() == 2) {
                return result.get(1);
            }
            return null;
        }
    }

    /**
     * 向列表左侧批量插入数据，并只保留最多N条
     *
     * @param key      redis key
     * @param maxCount 最大保留条数
     * @param values   要插入的值（可变参数）
     */
    public void lpushAndTrim(String key, int maxCount, String... values) {

        
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.lpush(key, values);
            jedis.ltrim(key, 0, maxCount - 1);
        }
    }

    /**
     * 获取 Redis 列表全部元素
     *
     * @param key redis key
     * @return 列表所有元素（从左到右，最新到最旧）
     */
    public List<String> lgetAll(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.lrange(key, 0, -1);
        }
    }


    public String lindex(String key, long index) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.lindex(key, index);
        }
    }

    /**
     * 获取 Redis 列表全部元素，顺序为最旧到最新
     *
     * @param key redis key
     * @return 列表所有元素（最旧到最新）
     */
    public List<String> lgetAllOldestFirst(String key) {
        try (Jedis jedis = jedisPool.getResource()) {
            List<String> list = jedis.lrange(key, 0, -1);
            java.util.Collections.reverse(list);
            return list;
        }
    }

    // Sorted Set
    public Long zadd(String key, double score, String member) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.zadd(key, score, member);
        }
    }

    public List<String> zrange(String key, long start, long end) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.zrange(key, start, end);
        }
    }

    /**
     * 获取 zset 区间（按分数从高到低，包含相等，返回列表）
     */
    public List<String> zrevrangeByScore(String key, double max, double min) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.zrevrangeByScore(key, max, min);
        }
    }

    /**
     * 获取 zset 区间，带分页（按分数从高到低，offset, count）
     */
    public List<String> zrevrangeByScore(String key, double max, double min, int offset, int count) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.zrevrangeByScore(key, max, min, offset, count);
        }
    }

    public Long incrBy(String key, long value) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.incrBy(key, value);
        }
    }

    /**
     * set key value with NX/XX and EX/PX options
     *
     * @param key    redis key
     * @param value  redis value
     * @param params SetParams (如 SetParams.setParams().nx().ex(30))
     * @return OK if set, null if not set
     */
    public String set(String key, String value, redis.clients.jedis.params.SetParams params) {
        try (Jedis jedis = jedisPool.getResource()) {
            return jedis.set(key, value, params);
        }
    }


    /**
     * ✅ 启动一个订阅线程（支持多个频道）
     *
     * @param listener  JedisPubSub 实现（回调）
     * @param channels  要订阅的频道（一个或多个）
     */
    public void asyncSubscribe(JedisPubSub listener, String... channels) {
        new Thread(() -> {
            while (true) {
                try (Jedis jedis = jedisPool.getResource()) {
                    logger.info("📡 Redis 订阅中: {} ",String.join(",", channels));
                    jedis.subscribe(listener, channels); // ⚠️ 阻塞
                } catch (Exception e) {
                    logger.error("❌ Redis subscribe 连接丢失，5s后重连: " + e.getMessage());
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException ex) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }, "redis-sub-thread").start();
    }

    /**
     * ✅ 支持 Pattern 订阅（通配符）
     *
     * @param listener  JedisPubSub 实现
     * @param patterns  支持通配符的频道
     */
    public void asyncPSubscribe(JedisPubSub listener, String... patterns) {
        new Thread(() -> {
            while (true) {
                try (Jedis jedis = jedisPool.getResource()) {
                    System.out.println("📡 Redis Pattern 订阅: " + String.join(",", patterns));
                    jedis.psubscribe(listener, patterns);
                } catch (Exception e) {
                    System.err.println("❌ Redis psubscribe 连接丢失，5s后重连: " + e.getMessage());
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException ex) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }, "redis-psub-thread").start();
    }

    /**
     * ✅ 主动取消订阅
     */
    public void unsubscribe(JedisPubSub listener, String... channels) {
        listener.unsubscribe(channels);
    }

    /**
     * ✅ 主动取消 pattern 订阅
     */
    public void punsubscribe(JedisPubSub listener, String... patterns) {
        listener.punsubscribe(patterns);
    }
}
